/*
 *   Copyright 1999-2016 Asiainfo Technologies(China),Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.asiainfo.redis.service.clusterhandler;

import com.asiainfo.redis.entity.Cluster;
import com.asiainfo.redis.entity.ClusterNode;
import com.asiainfo.redis.repository.ClusterNodeRepository;
import com.asiainfo.redis.repository.ClusterRepository;
import com.asiainfo.redis.service.redis.MyJedisConnectionFactory;
import com.asiainfo.redis.utils.SpringContextUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.ClusterStateFailureException;
import org.springframework.data.redis.connection.RedisClusterCommands;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Protocol;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @version v 1.0 on 2016/5/9 15:54
 * @auther william.xu
 */
//@Component
public class RedisClusterHandler {

    protected static final Logger LOGGER = LoggerFactory.getLogger(RedisClusterHandler.class);

    //@Resource
    //private ClusterNodeRepository clusterNodeRepository;

    private RedisClusterConfiguration redisClusterConfiguration;

    private final static int TOTAL_SLOT = 16384;
    private final static int WAIT_A_MINUTE = 500;
    private final static int CHECK_COUNTS = 20;

    public RedisClusterHandler(){

    }

    public RedisClusterHandler(Long clusterId) throws Exception{
        this.setRedisClusterConfiguration(clusterId);
    }

    public RedisClusterHandler(List<RedisClusterNode> nodeList) throws Exception{
        this.setRedisClusterConfiguration(nodeList);
    }

    public <T> T execute(RedisClusterCallback<T> callback) throws Exception {
        Assert.notNull(callback, "ClusterCallback must not be null!");
        RedisClusterConnection connection = getRedisClusterConnection();
        try {
            return callback.doInRedisCluster(connection);
        } finally {
            //connection.close();
            closeRedisClusterConnection(connection);
        }
    }

    public void meet() {
        //redis-cli到其中一个节点，进行meet操作
        Set<RedisNode> redisNodes = redisClusterConfiguration.getClusterNodes();
        RedisNode redisNode = redisNodes.iterator().next();
        Jedis jedis = null;
        try {
            jedis = new Jedis(redisNode.getHost(), redisNode.getPort(), Protocol.DEFAULT_TIMEOUT);
            Iterator<RedisNode> it = redisNodes.iterator();
            while (it.hasNext()) {
                RedisNode node = it.next();
                jedis.clusterMeet(node.getHost(), node.getPort());
            }
        } finally {
            jedis.close();
        }
    }

    public void meet(ClusterNode existNode, List<ClusterNode> masterNodes) {
        //redis-cli到existNode，进行meet操作
        Jedis jedis = null;
        try {
            jedis = new Jedis(existNode.getIp(), existNode.getPort(), Protocol.DEFAULT_TIMEOUT);
            for (ClusterNode masterNode : masterNodes) {
                jedis.clusterMeet(masterNode.getIp(), masterNode.getPort());
                List<ClusterNode> slaveNodes = masterNode.getSlaveNodes();
                for (ClusterNode slaveNode : slaveNodes) {
                    jedis.clusterMeet(slaveNode.getIp(), slaveNode.getPort());
                }
            }
        } finally {
            jedis.close();
        }
    }


    public void checkAllMeeting(RedisClusterConnection redisClusterConnection) {
        int count = 0;
        while (true) {
            LOGGER.info("waiting cluster join...");
            try {
                Thread.sleep(WAIT_A_MINUTE);
            } catch (Exception e) {
            }
            Set<RedisNode> configNodes = redisClusterConfiguration.getClusterNodes();
            JedisCluster cluster = (JedisCluster) redisClusterConnection.getNativeConnection();
            boolean meetAll = true;
            for (Map.Entry<String, JedisPool> entry : cluster.getClusterNodes().entrySet()) {
                Jedis jedis = null;
                try {
                    jedis = entry.getValue().getResource();
                    Set<RedisClusterNode> nodes = Converters.toSetOfRedisClusterNodes(jedis.clusterNodes());
                    if (nodes.size() < configNodes.size()) {
                        meetAll = false;
                        break;
                    }
                } catch (Exception e) {
                    LOGGER.error(entry.getKey() + " error:", e);
                    meetAll = false;
                } finally {
                    if (jedis != null) {
                        jedis.close();
                    }
                }
            }
            count++;
            if (count > CHECK_COUNTS) {
                break;
            }
            if (meetAll) {
                try {
                    //JedisClusterConnection.getTopology()每隔100ms获取一次cluster nodes，保证获取最新
                    Thread.sleep(WAIT_A_MINUTE);
                } catch (Exception e) {
                }
                break;
            }
//            Set<RedisClusterNode> clusterNodes = ((JedisClusterConnection)redisClusterConnection).clusterGetNodes();
//            if (configNodes.size() == clusterNodes.size()){
//                break;
//            }
        }

    }

    public void replicate(List<ClusterNode> masterNodes, RedisClusterConnection redisClusterConnection) {
        for (ClusterNode masterNode : masterNodes) {
            List<ClusterNode> slaveNodes = masterNode.getSlaveNodes();
            for (ClusterNode slaveNode : slaveNodes) {
                RedisClusterNode master = new RedisClusterNode(masterNode.getIp(), masterNode.getPort());
                RedisClusterNode slave = new RedisClusterNode(slaveNode.getIp(), slaveNode.getPort());
                redisClusterConnection.clusterReplicate(master, slave);
            }
        }
    }

    public void addSlots(List<ClusterNode> masterNodes, RedisClusterConnection redisClusterConnection) {
        int masterNodeSize = masterNodes.size();
        int perNodeCount = TOTAL_SLOT / masterNodeSize;
        for (int i = 0; i < masterNodeSize; i++) {
            ClusterNode masterNode = masterNodes.get(i);
            RedisClusterNode redisClusterNode = new RedisClusterNode(masterNode.getIp(), masterNode.getPort());
            int lowerBound = i * perNodeCount;
            int upperBound = i * perNodeCount + perNodeCount - 1;
            if (i == (masterNodeSize - 1)) {
                upperBound = TOTAL_SLOT - 1;
            }
            RedisClusterNode.SlotRange slotRange = new RedisClusterNode.SlotRange(lowerBound, upperBound);
            redisClusterConnection.clusterAddSlots(redisClusterNode, slotRange);
        }

    }

    public void forgetNode(RedisClusterConnection redisClusterConnection, RedisClusterNode removeNode) {
//        JedisCluster cluster = (JedisCluster)redisClusterConnection.getNativeConnection();
//        for (Map.Entry<String, JedisPool> entry : cluster.getClusterNodes().entrySet()) {
//            Jedis jedis = null;
//            try {
//                jedis = entry.getValue().getResource();
//                if (entry.getKey().equals(removeNode.toString())){
//                    //对节点停止服务
//                    //redisClusterConnection.shutdown(removeNode);
//                    jedis.shutdown();
//                }else {
//                    jedis.clusterForget(removeNode.getId());
//                }
//            }catch (Exception e){
//                LOGGER.warn(entry.getKey() + " error:", e);
//            } finally {
//                if (jedis != null) {
//                    jedis.close();
//                }
//            }
//        }
        Iterator<RedisClusterNode> it = redisClusterConnection.clusterGetNodes().iterator();
        while (it.hasNext()) {
            RedisClusterNode redisClusterNode = it.next();
            Jedis jedis = null;
            try {
                jedis = new Jedis(redisClusterNode.getHost(), redisClusterNode.getPort(), Protocol.DEFAULT_TIMEOUT);
                if (redisClusterNode.toString().equals(removeNode.toString())) {
                    jedis.shutdown();
                } else {
                    jedis.clusterForget(removeNode.getId());
                }
            } catch (Exception e) {
                LOGGER.warn(redisClusterNode + " error:", e);
            } finally {
                if (jedis != null) {
                    jedis.close();
                }
            }
        }

    }

    public boolean checkIfMasterNode(RedisClusterNode node, RedisClusterConnection redisClusterConnection) {
        boolean result = false;
        Iterator<RedisClusterNode> it = redisClusterConnection.clusterGetNodes().iterator();
        while (it.hasNext()) {
            RedisClusterNode redisClusterNode = it.next();
            if (redisClusterNode.toString().equals(node.toString())) {
                if (redisClusterNode.isMaster()) {
                    result = true;
                } else {
                    result = false;
                }
                break;
            }
        }
        return result;
    }

    public RedisClusterNode lookupNode(RedisClusterNode node, RedisClusterConnection redisClusterConnection) throws
            Exception {
        Iterator<RedisClusterNode> iterator = redisClusterConnection.clusterGetNodes().iterator();
        while (iterator.hasNext()) {
            RedisClusterNode redisClusterNode = iterator.next();
            if (redisClusterNode.toString().equals(node.toString())) {
                return redisClusterNode;
            }
        }
        throw new Exception("找不到节点" + node.toString());
    }

    public void migrateSlot(int slot, RedisClusterNode fromRedisClusterNode, RedisClusterNode toRedisClusterNode,
                            RedisClusterConnection redisClusterConnection) throws Exception {
        boolean isMasterNode = checkIfMasterNode(fromRedisClusterNode, redisClusterConnection);
        Assert.isTrue(isMasterNode, "迁移的源节点" + fromRedisClusterNode.getHost() + ":" + fromRedisClusterNode.getPort()
                + "不是主节点");
        isMasterNode = checkIfMasterNode(toRedisClusterNode, redisClusterConnection);
        Assert.isTrue(isMasterNode, "迁移的目标节点" + toRedisClusterNode.getHost() + ":" + toRedisClusterNode.getPort() +
                "不是主节点");
        //有问题不能直接用
        //redisClusterConnection.clusterSetSlot(redisClusterNode, slot, );
        //目标节点上置导入状态
        redisClusterConnection.clusterSetSlot(toRedisClusterNode, fromRedisClusterNode, slot, RedisClusterCommands
                .AddSlots.IMPORTING);
        //源节点上置迁移状态
        redisClusterConnection.clusterSetSlot(fromRedisClusterNode, toRedisClusterNode, slot, RedisClusterCommands
                .AddSlots.MIGRATING);
        //取出slot的key的数量
        long countKeys = redisClusterConnection.clusterCountKeysInSlot(slot);
        //取出slot的key
        List<byte[]> keyList = redisClusterConnection.clusterGetKeysInSlot(slot, (int) countKeys);
        LOGGER.info("正在迁移slot=" + slot + ", key size=" + keyList.size());
        for (byte[] key : keyList) {
            RedisNode src = new RedisNode(fromRedisClusterNode.getHost(), fromRedisClusterNode.getPort());
            RedisNode target = new RedisNode(toRedisClusterNode.getHost(), toRedisClusterNode.getPort());
            try {
                redisClusterConnection.migrate(key, src, target, 0, RedisServerCommands.MigrateOption.REPLACE);
            } catch (Exception e) {
                if (e.getMessage().contains("BUSYKEY")) {
                    LOGGER.warn("目标节点key已经存在，" + e.toString());
                } else {
                    throw e;
                }
            }
        }
//        //源节点上指派目标槽位操作
//        //当源节点最后一个slot被迁移后，以下指令有时会抛ERR Please use SETSLOT only with masters错误
//        redisClusterConnection.clusterSetSlot(fromRedisClusterNode, toRedisClusterNode, slot, RedisClusterCommands
// .AddSlots.NODE);
//        //目标节点上指派目标节点槽位操作
//        redisClusterConnection.clusterSetSlot(toRedisClusterNode, toRedisClusterNode, slot, RedisClusterCommands
// .AddSlots.NODE);
        //在集群中所有的主节点上指派目标节点槽位操作
        Iterator<RedisClusterNode> it = redisClusterConnection.clusterGetNodes().iterator();
        while (it.hasNext()) {
            RedisClusterNode redisClusterNode = it.next();
            if (redisClusterNode.isMaster()) {
                redisClusterConnection.clusterSetSlot(redisClusterNode, toRedisClusterNode, slot,
                        RedisClusterCommands.AddSlots.NODE);
            }
        }

    }


    public Cluster getClusterAndNodesInfo(Long clusterId, RedisClusterConnection connection) {
        Cluster cluster = new Cluster();
        ClusterRepository clusterRepository = SpringContextUtil.getBean(ClusterRepository.class);
        ClusterNodeRepository clusterNodeRepository = SpringContextUtil.getBean(ClusterNodeRepository.class);
        try {
            cluster = clusterRepository.findOne(clusterId);
            Assert.notNull(cluster, "找不到该集群！");

            List<ClusterNode> masterNodes = new ArrayList<ClusterNode>();
            Map<String, ClusterNode> masterNodesMap = new HashMap<String, ClusterNode>();
            Iterator<RedisClusterNode> iterator = connection.clusterGetNodes().iterator();
            while (iterator.hasNext()) {
                RedisClusterNode redisClusterNode = iterator.next();

                String host = redisClusterNode.getHost();
                Integer port = redisClusterNode.getPort();
                List<ClusterNode> list = clusterNodeRepository.findByClusterIdAndIpAndPort(clusterId, host, port);
                Long id = null;
                if (list.size() > 0) {
                    id = list.get(0).getId();
                }
                String nodeId = redisClusterNode.getId();
                String masterNodeId = redisClusterNode.getMasterId();
                if (masterNodeId == null) {
                    //说明当前节点是主节点
                    ClusterNode masterNode = masterNodesMap.get(nodeId);
                    if (masterNode == null) {
                        masterNode = new ClusterNode();
                        masterNode.setSlaveNodesMap(new HashMap<String, ClusterNode>());
                        masterNodesMap.put(nodeId, masterNode);
                    }
                    masterNode.setId(id);
                    masterNode.setNodeId(nodeId);
                    masterNode.setIp(host);
                    masterNode.setPort(port);
                    masterNode.setState(redisClusterNode.getLinkState().toString());
                    //masterNode.setSlots(redisClusterNode.getSlotRange().getSlots());//数据量太多，暂时不返回到客户端了
                    masterNode.setSlotsLength(redisClusterNode.getSlotRange().getSlots().size());
                } else {
                    //说明当前节点时从节点,根据masterNodeId获取主节点
                    ClusterNode masterNode = masterNodesMap.get(masterNodeId);
                    if (masterNode == null) {
                        masterNode = new ClusterNode();
                        masterNode.setSlaveNodesMap(new HashMap<String, ClusterNode>());
                        masterNodesMap.put(masterNodeId, masterNode);
                    }
                    ClusterNode slaveNode = new ClusterNode();
                    slaveNode.setId(id);
                    slaveNode.setNodeId(nodeId);
                    slaveNode.setIp(host);
                    slaveNode.setPort(port);
                    slaveNode.setState(redisClusterNode.getLinkState().toString());
                    masterNode.getSlaveNodesMap().put(nodeId, slaveNode);
                }
            }
            masterNodes.addAll(masterNodesMap.values());
            cluster.setMasterNodes(masterNodes);
            cluster.setClusterInfo(connection.clusterGetClusterInfo());
            cluster.setState(cluster.getClusterInfo().getState());

            //获取cluster-slave-no-failover配置信息
            for(ClusterNode masterNode : masterNodes){
                setClusterSlaveNoFailoverInfo(masterNode);
                Map<String, ClusterNode> map = masterNode.getSlaveNodesMap();
                Collection<ClusterNode> slaveNodes = map.values();
                for(ClusterNode slaveNode : slaveNodes){
                    setClusterSlaveNoFailoverInfo(slaveNode);
                }
            }

        } catch (ClusterStateFailureException e) {
            LOGGER.warn(e.toString());
            //集群中所有节点都访问不了
            List<ClusterNode> clusterNodeList = clusterNodeRepository.findByClusterId(clusterId);
            for (ClusterNode node : clusterNodeList) {
                node.setState("DISCONNECTED");
            }
            cluster.setMasterNodes(clusterNodeList);
            cluster.setState("shutdown");
        }
        return cluster;
    }

    private void setClusterSlaveNoFailoverInfo(ClusterNode node){
        if (node.getState() != null && node.getState().equals(RedisClusterNode.LinkState
                .CONNECTED.toString())){
            Jedis jedis = null;
            try {
                jedis = new Jedis(node.getIp(), node.getPort(), Protocol.DEFAULT_TIMEOUT);
                List<String> result = jedis.configGet("cluster-slave-no-failover");
                if (result != null && result.size() > 1){
                    node.setNoFailover(result.get(1));
                }
            }finally {
                if (jedis != null){
                    jedis.close();
                }
            }
        }
    }

    private void setRedisClusterConfiguration(List<RedisClusterNode> nodeList) throws Exception {
        RedisClusterConfiguration configuration = new RedisClusterConfiguration();
        configuration.setMaxRedirects(nodeList.size() + 1);
        for (RedisClusterNode clusterNode : nodeList) {
            configuration.clusterNode(clusterNode.getHost(), clusterNode.getPort());
        }
        this.redisClusterConfiguration = configuration;
    }


    private void setRedisClusterConfiguration(Long clusterId) throws Exception {
        this.redisClusterConfiguration = getRedisClusterConfiguration(clusterId);
    }

    private static RedisClusterConfiguration getRedisClusterConfiguration(Long clusterId) throws Exception{
        ClusterNodeRepository clusterNodeRepository = SpringContextUtil.getBean(ClusterNodeRepository.class);
        List<ClusterNode> clusterNodeList = clusterNodeRepository.findByClusterId(clusterId);
        if (clusterNodeList.size() == 0) {
            throw new Exception("找不到该集群的任何节点");
        }
        RedisClusterConfiguration configuration = new RedisClusterConfiguration();
        configuration.setMaxRedirects(clusterNodeList.size() + 1);
        for (ClusterNode clusterNode : clusterNodeList) {
            configuration.clusterNode(clusterNode.getIp(), clusterNode.getPort());
        }
        return configuration;
    }

    private static RedisClusterConnection getRedisClusterConnection(RedisClusterConfiguration configuration){
        JedisConnectionFactory jedisConnectionFactory = new MyJedisConnectionFactory(configuration);
        jedisConnectionFactory.setTimeout(Protocol.DEFAULT_TIMEOUT + 3000);
        jedisConnectionFactory.afterPropertiesSet();
        return jedisConnectionFactory.getClusterConnection();
    }


    private RedisClusterConnection getRedisClusterConnection() {
        return getRedisClusterConnection(this.redisClusterConfiguration);
    }

    public static RedisClusterConnection getRedisClusterConnection(Long clusterId) throws Exception{
        RedisClusterConfiguration configuration = getRedisClusterConfiguration(clusterId);
        return getRedisClusterConnection(configuration);
    }


    public static void closeRedisClusterConnection(RedisClusterConnection connection) {
        try {
            ((JedisCluster) connection.getNativeConnection()).close();
        } catch (IOException e) {
        }
    }

    private boolean checkAllClusterNodesConnected(RedisClusterConnection redisClusterConnection) throws Exception {
        Iterator<RedisClusterNode> iterator = redisClusterConnection.clusterGetNodes().iterator();
        while (iterator.hasNext()) {
            RedisClusterNode redisClusterNode = iterator.next();
            if (redisClusterNode.getLinkState() == RedisClusterNode.LinkState.DISCONNECTED) {
                return false;
            }
        }
        return true;
    }

    public void checkClusterNodes(RedisClusterConnection redisClusterConnection) throws Exception {
        if (!checkAllClusterNodesConnected(redisClusterConnection)) {
            throw new Exception("该集群存在部分节点连接不正常，不能执行此操作");
        }
    }

    /**
     * 校验srcNode的cluster nodes信息里面是否包含targetNode的信息
     *
     * @param srcNode
     * @param targetNode
     */
    private void checkMeeting(RedisClusterNode srcNode, RedisClusterNode targetNode) {
        Jedis src = new Jedis(srcNode.getHost(), srcNode.getPort(), Protocol.DEFAULT_TIMEOUT);
        boolean meetSuccess = false;
        while (!meetSuccess) {
            String nodes = src.clusterNodes();
            if (nodes.contains(targetNode.toString())) {
                meetSuccess = true;
            }
            if (!meetSuccess) {
                LOGGER.info(targetNode + " wait " + srcNode + " meet success ...");
                try {
                    Thread.sleep(WAIT_A_MINUTE);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        src.close();

    }

}
